package com.service.socket.ws;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.service.socket.common.entity.ReceiveMessageVo;
import com.service.socket.common.entity.SendMessageVo;
import com.service.socket.ws.entity.WebSocketSender;
import com.service.socket.common.enums.MsgSendType;
import com.service.socket.common.enums.SubTheme;
import com.service.socket.support.TradingSpotCache;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @ServerEndpoint 注解是一个类层次的注解，它的功能主要是将目前的类定义成一个websocket服务器端,
 * 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
 */
@Slf4j
@Component
@ServerEndpoint(value = "/ws")
public class TradingSpotWS {
    // 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount;

    //实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key为用户标识
    private static Map<String, Map<String, WebSocketSender>> subThemeMap = new ConcurrentHashMap<>();

    private static Map<String, Session> connectionMap = new ConcurrentHashMap<>();

    // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    private String id;

    public static TradingSpotCache spotCache;

    /**
     * 连接建立成功调用的方法
     *
     * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @OnOpen
    public void onOpen(Session session) {
        log.info("session:{}", session);
        session.setMaxTextMessageBufferSize(104857600);//100M
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (StringUtils.isNotBlank(id)) {
            log.info("会员<账号:{}>关闭连接", id);
            for (String key : subThemeMap.keySet()) {
                if (!ObjectUtils.isEmpty(subThemeMap.get(key))) {
                    subThemeMap.get(key).remove(id);
                }
            }
            connectionMap.remove(this.id);
        }
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     * @param session 可选的参数
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("receive msg:{}", message);
        ReceiveMessageVo messageVo = JSON.parseObject(message, ReceiveMessageVo.class);
        SendMessageVo.SendMessageVoBuilder builder = SendMessageVo.builder();

        if (StringUtils.isNotBlank(messageVo.getPong())) {
            log.info("websocket pong:{}", messageVo.getPong());
            return;
        }

        if (StringUtils.isNotBlank(messageVo.getSub())) {
            builder.type(MsgSendType.subbed.getMsg()).data(true);
            //todo：添加相应的订阅频道
            Map<String, WebSocketSender> themeMap = subThemeMap.get(messageVo.getSub());
            if (!subThemeMap.containsKey(messageVo.getSub())) {
                themeMap = Maps.newConcurrentMap();
                subThemeMap.put(messageVo.getSub(), themeMap);
            }
            this.id = messageVo.getParam();
            connectionMap.put(this.id, session);
            themeMap.put(messageVo.getParam(), WebSocketSender.builder()
                    .channel(messageVo.getSub()).session(session).build());

            //todo:发送相应订阅频道的数据
            this.sendSuccessSubMsg(messageVo.getSub(), session);

            log.info("会员<账号:{}>订阅频道:{},当前频道人数:{}", messageVo.getParam(), messageVo.getSub(),
                    themeMap.size());
            return;
        }

        if (StringUtils.isNotBlank(messageVo.getUnsub())) {
            builder.type(MsgSendType.unsubbed.getMsg()).data(true).status(true);
            this.sendSingleMsg(session, builder.build());
            //todo：从相应频道删除
            Map<String, WebSocketSender> themeMap = subThemeMap.get(messageVo.getUnsub());
            if (ObjectUtils.isEmpty(themeMap)) {
                return;
            }
            if (StringUtils.isNotBlank(messageVo.getParam())) {
                log.info("会员<账号:{}>取消订阅频道:{},当前频道人数:{}", messageVo.getParam(), messageVo.getSub(),
                        themeMap.size());
                themeMap.remove(messageVo.getParam());
            }
        }
    }

    public void sendSuccessSubMsg(String subMsg, Session session) {
        if (StringUtils.isBlank(subMsg)) {
            return;
        }
        if (!subMsg.contains(":")) {
            return;
        }
        String[] subArr = subMsg.split(":");
        SubTheme subTheme = SubTheme.valueOf(subArr[0]);
        if (subArr.length <= 1 || ObjectUtils.isEmpty(subTheme)) {
            return;
        }
        SendMessageVo.SendMessageVoBuilder builder = SendMessageVo.builder();
        boolean isSend = false;
        switch (subTheme) {
            case kline:
                builder.type(MsgSendType.kline.getMsg())
                        .status(true)
                        .data(TradingSpotWS.spotCache.getKline(subArr[1], Integer.parseInt(subArr[2])));
                break;
            case kline_statistics:
                builder.type(MsgSendType.kline_statistics.getMsg())
                        .status(true)
                        .data(TradingSpotWS.spotCache.getKlineStatistic(subArr[1]));
                break;
            case handicap:
                builder.type(MsgSendType.handicap.getMsg())
                        .status(true)
                        .data(TradingSpotWS.spotCache.getDepth(subArr[1]));
                break;
            case spot_trade:
                builder.status(true).type(MsgSendType.subbed.getMsg()).data(true);
                this.sendSingleMsg(session, builder.build());
                isSend = true;
                break;
        }
        if (isSend) {
            return;
        }
        try {
            session.getBasicRemote().sendText(JSON.toJSONString(builder.build()));
        } catch (Exception e) {
            log.error("send msg fail");
        }
    }

    /**
     * 发生错误时调用
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误:{}", error.getMessage());
//        error.printStackTrace();
    }

    public void sendSingleMsg(Session session, SendMessageVo sendMessageVo) {
        if (ObjectUtils.isEmpty(sendMessageVo)) {
            return;
        }
        try {
            session.getBasicRemote().sendText(JSON.toJSONString(sendMessageVo));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void sendSingleMsg(MsgSendType type, String userAccount, SendMessageVo sendMessageVo) {
        String channel = this.buildChannel(type, sendMessageVo.getSymbol(), sendMessageVo.getPeriod());
        Map<String, WebSocketSender> themeMap = subThemeMap.get(channel);
        if (ObjectUtils.isEmpty(themeMap)) {
            return;
        }
        sendMessageVo.setType(type.getMsg());
        log.info("频道:{}中会员:{}发送消息:{}", channel, userAccount, sendMessageVo);
        if (themeMap.containsKey(userAccount)) {
            WebSocketSender sender = themeMap.get(userAccount);
            if (sender.getSession().isOpen()) {
                sender.sendData(sendMessageVo);
            } else {
                themeMap.remove(userAccount);
            }
        }
    }

    //广播发送
    public void broadcastThemeMsg(MsgSendType type, SendMessageVo sendMessageVo) {
        String channel = this.buildChannel(type, sendMessageVo.getSymbol(), sendMessageVo.getPeriod());
        Map<String, WebSocketSender> themeMap = subThemeMap.get(channel);
        if (ObjectUtils.isEmpty(themeMap)) {
            return;
        }
        sendMessageVo.setType(type.getMsg());
        log.info("频道:{}广播消息", channel);
        for (Iterator<Map.Entry<String, WebSocketSender>> it = themeMap.entrySet().iterator(); it.hasNext(); ) {
            Map.Entry<String, WebSocketSender> item = it.next();
            WebSocketSender sender = item.getValue();
            if (ObjectUtils.isEmpty(sender)) {
                continue;
            }
            if (!sender.getSession().isOpen()) {
                it.remove();
                continue;
            }
            sender.sendData(sendMessageVo);
        }
    }

    //广播发送ping
    public void broadcastPingMsg(SendMessageVo sendMessageVo) {
        for (Iterator<Map.Entry<String, Session>> it = connectionMap.entrySet().iterator(); it.hasNext(); ) {
            Map.Entry<String, Session> item = it.next();
            Session session = item.getValue();
            if (!session.isOpen()) {
                it.remove();
                continue;
            }
            sendMessage(session, JSON.toJSONString(sendMessageVo));
        }
    }

    /**
     * 发送消息需注意方法加锁synchronized，避免阻塞报错
     * 注意session.getBasicRemote()与session.getAsyncRemote()的区别
     *
     * @param message
     */
    private void sendMessage(Session session, String message) {
        if(session.isOpen()){
            try {
                session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    private String buildChannel(MsgSendType type, String symbol, Integer period) {
        if (StringUtils.isNotBlank(symbol) && symbol.contains("/") && !symbol.contains("_")) {
            symbol = symbol.replace("/", "_");
        }
        return ObjectUtils.isEmpty(period) ? type.getMsg() + ":" + symbol : type.getMsg() + ":" + symbol + ":" + period;
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        TradingSpotWS.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        if (TradingSpotWS.onlineCount > 0) {
            TradingSpotWS.onlineCount--;
        }
    }
}